package com.bw.func;

import com.bw.been.OrderWide;
import com.bw.util.ConfigUtil;
import com.bw.util.HBaseUtil;
import com.bw.util.SourceHbaseUntil;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.async.ResultFuture;
import org.apache.flink.streaming.api.functions.async.RichAsyncFunction;
import org.apache.hadoop.hbase.client.Connection;

import java.util.Collections;

public class AsyncDatabaseRequest extends RichAsyncFunction<String, OrderWide> {

    /** The database specific client that can issue concurrent requests with callbacks */
    private  static Connection connection = null;

    private String tableName;

    public AsyncDatabaseRequest(String tableName) {
        this.tableName = tableName;
    }

    @Override
    public void open(Configuration parameters) throws Exception {
        connection = HBaseUtil.getConnection(
                ConfigUtil.getZookeeperQuorum()
                , ConfigUtil.getZookeeperClientPort());
        System.out.println("初始化hbase链接>>>>>>>>>");
    }

    @Override
    public void close() throws Exception {
        connection.close();
    }

    @Override
    public void asyncInvoke(String key, final ResultFuture<OrderWide> resultFuture) throws Exception {
        String str = SourceHbaseUntil.sc(tableName, key);
//        resultFuture.complete(Collections.singleton(str));
    }
}
